package org.atlantis.ymer.server.transport;

import java.util.LinkedList;
import java.util.List;

import org.atlantis.ymer.DataForwarder;
import org.atlantis.ymer.WebSession;
import org.atlantis.ymer.command.ConnectCommand;
import org.atlantis.ymer.command.DisconnectCommand;
import org.atlantis.ymer.command.JsonCommand;
import org.atlantis.ymer.command.PingCommand;
import org.atlantis.ymer.command.PubCommand;
import org.atlantis.ymer.common.HttpResponseUtils;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.handler.codec.http.HttpHeaders;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpResponse;
import org.jboss.netty.util.CharsetUtil;

public class LongPollingTransport extends HttpTransport {

	@Override
	protected void handleGetMethod(ChannelHandlerContext ctx, HttpRequest request, RequestParameter param) {
		String webSessionId = param.getWebSessionId();
		boolean jsonp = param.getString("jsonp") != null;
		if (webSessionId == null) {
			connect(ctx, request, jsonp);
			return;
		}
		WebSession webSession = tryGetWebSession(ctx, webSessionId);
		if (webSession == null) return;
		LongPollingDataForwarder forwarder = (LongPollingDataForwarder) webSession.getBindedForwarder();
		forwarder.resetCtx(ctx);
		forwarder.connect(HttpHeaders.isKeepAlive(request), jsonp);
	}

	@Override
	protected void handlePostMethod(ChannelHandlerContext ctx, HttpRequest request, RequestParameter param) {
		String webSessionId = param.getWebSessionId();
		WebSession webSession = tryGetWebSession(ctx, webSessionId);
		if (webSession == null) return;
		passContentToWebSession(param.getContent(), webSession);
		ctx.getChannel().write(HttpResponseUtils.createCrossDomainHeaderResponse()).addListener(ChannelFutureListener.CLOSE);
	}

	private void connect(ChannelHandlerContext ctx, HttpRequest request, boolean jsonp) {
		WebSession webSession = createNewWebSession();
		DataForwarder dataForwarder = new LongPollingDataForwarder(ctx, webSession, HttpHeaders.isKeepAlive(request), jsonp);
		webSession.onData(new ConnectCommand(dataForwarder));
	}

	private class LongPollingDataForwarder extends AbstractDataForwarder {

		private boolean waiting;
		private boolean isKeepAlive;
		private boolean jsonp;
		/*
		 * 需要同步，因为客户端通过connect读取数据和服务端通过sendMessage、sendPing等发送数据不是相同线程，
		 * 所以会并发对dataFrameBuffers操作。因此必须对dataFrameBuffers同步。
		 */
		private List<JsonCommand> commandBuffers = new LinkedList<JsonCommand>();

		protected LongPollingDataForwarder(ChannelHandlerContext ctx, WebSession webSession, boolean isKeepAlive,
				boolean jsonp) {
			super(ctx, webSession, false);
			connect(isKeepAlive, jsonp);
		}

		void connect(boolean isKeepAlive, boolean jsonp) {
			this.waiting = true;
			this.isKeepAlive = isKeepAlive;
			this.jsonp = jsonp;
			String jsonContent = null;
			synchronized (commandBuffers) {
				if (commandBuffers.isEmpty()) return;
				jsonContent = JsonCommand.parseCommands(this.commandBuffers);
				commandBuffers.clear();
			}
			send(jsonContent);
		}

		@Override
		public void sendConnect() {
			send(wrapToArray(new ConnectCommand(webSession.getId()).getContent()));
		}

		@Override
		public void sendPing() {
			if (waiting) {
				send(wrapToArray(PingCommand.instance.getContent()));
				return;
			}
			synchronized (commandBuffers) {
				commandBuffers.add(PingCommand.instance);
			}
		}

		@Override
		public void sendMessage(PubCommand command) {
			if (waiting) {
				send(wrapToArray(command.getContent()));
				return;
			}
			synchronized (commandBuffers) {
				commandBuffers.add(command);
			}
		}

		@Override
		public void sendDisconnect(DisconnectCommand command) {
			isKeepAlive = false;
			if (waiting) {
				send(wrapToArray(command.getContent()));
				return;
			}
			synchronized (commandBuffers) {
				commandBuffers.add(command);
			}
		}

		private String wrapToArray(String jsonContent) {
			return new StringBuilder("[").append(jsonContent).append("]").toString();
		}

		private void send(String jsonContent) {
			waiting = false;
			Channel channel = getChannel();
			if (channel == null) return;

			if (jsonp) jsonContent = new StringBuilder().append("__ymer_c_b(").append(jsonContent).append(")").toString();

			HttpResponse response = buildResponse();
			response.setContent(ChannelBuffers.copiedBuffer(jsonContent, CharsetUtil.UTF_8));
			HttpHeaders.setContentLength(response, response.getContent().readableBytes());

			ChannelFuture future = channel.write(response);
			if (!isKeepAlive) future.addListener(ChannelFutureListener.CLOSE);
		}

		private HttpResponse buildResponse() {
			HttpResponse response = HttpResponseUtils.createCrossDomainHeaderResponse();
			String contentType = (jsonp ? HttpResponseUtils.CONTENT_TYPE_JAVASCRIPT : HttpResponseUtils.CONTENT_TYPE_JSON);
			response.setHeader(HttpHeaders.Names.CONTENT_TYPE, contentType);
			return response;
		}
	}
}
